"""
项目管理服务层
提供项目相关的业务逻辑处理
"""
from typing import Any, List, Optional, Dict, Tuple
from datetime import datetime, timedelta, date, timezone
from bson import ObjectId  # pyright: ignore[reportMissingImports]
from motor.motor_asyncio import AsyncIOMotorDatabase  # pyright: ignore[reportMissingImports]
import logging
import hashlib
import json
import asyncio

from app.models.project import (
    ProjectCreate,
    ProjectUpdate,
    ProjectMemberAdd,
    ProjectMemberUpdate,
    ProjectCategory,
    ProjectMemberRole
)
from app.utils.task_utils import calculate_project_progress
from app.utils.state_utils import (
    validate_state_code,
    get_completed_state_code,
    get_completed_state_codes,
    get_state_options,
    get_state_code_by_name_keywords,
)
from app.core.permissions import Permissions
from app.services.role_service import RoleService

logger = logging.getLogger(__name__)


class ProjectService:
    """项目服务类"""

    def __init__(self):
        # 仪表盘统计缓存
        self._dashboard_cache: Dict[str, Dict[str, Any]] = {}
        self._dashboard_cache_timestamps: Dict[str, datetime] = {}
        self._dashboard_cache_ttl = timedelta(minutes=5)  # 缓存5分钟
        self._dashboard_cache_lock = asyncio.Lock()

    def _to_object_id(self, value: Any, field_name: str = "ID") -> Optional[ObjectId]:
        """将值转换为ObjectId，如无效返回None"""
        if value is None:
            return None
        if isinstance(value, ObjectId):
            return value
        if isinstance(value, str) and ObjectId.is_valid(value):
            return ObjectId(value)
        return None

    def _ensure_object_id(self, value: Any, field_name: str = "ID") -> ObjectId:
        """确保值可转换为ObjectId，否则抛出异常"""
        obj_id = self._to_object_id(value, field_name)
        if obj_id is None:
            raise ValueError(f"Invalid {field_name}")
        return obj_id

    def build_query_by_permissions(
        self,
        current_user: dict,
        search: Optional[str] = None,
        status: Optional[str] = None,
        priority: Optional[str] = None,
        category: Optional[ProjectCategory] = None,
        created_by: Optional[str] = None,
        member_id: Optional[str] = None,
        is_archived: Optional[bool] = None
    ) -> Dict[str, Any]:
        """
        根据用户权限构建查询条件
        
        Args:
            current_user: 当前用户信息
            search: 搜索关键词
            status: 状态筛选
            priority: 优先级筛选
            category: 分类筛选
            created_by: 创建者筛选
            member_id: 成员筛选
            is_archived: 是否归档筛选
            
        Returns:
            查询条件字典
        """
        query = {}
        
        # 权限检查：如果用户没有读取所有项目的权限，只能看到自己相关的项目
        user_permissions = current_user.get("permissions", [])
        if Permissions.PROJECT_READ_ALL not in user_permissions and Permissions.ADMIN not in user_permissions:
            query["$or"] = [
                {"is_public": True},
                {"created_by": ObjectId(current_user.get("user_id"))},
                {"members.user_id": ObjectId(current_user.get("user_id"))}
            ]
        
        # 搜索条件
        if search:
            query["$and"] = query.get("$and", [])
            query["$and"].append({
                "$or": [
                    {"name": {"$regex": search, "$options": "i"}},
                    {"description": {"$regex": search, "$options": "i"}}
                ]
            })
        
        if status:
            query["status"] = status
        
        if priority:
            query["priority"] = priority
        
        if category:
            query["category"] = category
        
        if created_by:
            try:
                query["created_by"] = ObjectId(created_by)
            except Exception:
                raise ValueError("Invalid created_by ID format")
        
        if member_id:
            try:
                query["members.user_id"] = ObjectId(member_id)
            except Exception:
                raise ValueError("Invalid member_id format")
        
        # 归档状态筛选
        if is_archived is not None:
            query["is_archived"] = is_archived
        # 如果没有明确指定，默认只显示未归档的项目（除非用户明确要求查看归档项目）
        # 注意：这里不设置默认值，让调用方决定是否要包含归档项目
        
        return query

    async def validate_priority(
        self,
        db: AsyncIOMotorDatabase,
        priority: Optional[str]
    ) -> bool:
        """
        验证优先级值是否有效
        
        Args:
            db: 数据库连接
            priority: 优先级代码
            
        Returns:
            是否有效
        """
        if not priority:
            return True
        
        priority_doc = await db.priorities.find_one({
            "module_type": "project",
            "code": priority,
            "is_enabled": True
        })
        return priority_doc is not None

    async def get_projects_list(
        self,
        db: AsyncIOMotorDatabase,
        current_user: dict,
        skip: int = 0,
        limit: int = 10,
        search: Optional[str] = None,
        status: Optional[str] = None,
        priority: Optional[str] = None,
        category: Optional[ProjectCategory] = None,
        created_by: Optional[str] = None,
        member_id: Optional[str] = None,
        is_archived: Optional[bool] = None
    ) -> tuple[List[Dict[str, Any]], int]:
        """
        获取项目列表
        
        Returns:
            (项目列表, 总数)
        """
        # 验证状态
        if status:
            is_valid = await validate_state_code(db, "project", status)
            if not is_valid:
                raise ValueError(f"Invalid status value: {status}")
        
        # 验证优先级
        if priority:
            if priority == "":
                priority = None
            else:
                is_valid = await self.validate_priority(db, priority)
                if not is_valid:
                    raise ValueError(f"Invalid priority value: {priority}")
        
        # 构建查询条件
        query = self.build_query_by_permissions(
            current_user,
            search,
            status,
            priority,
            category,
            created_by,
            member_id,
            is_archived
        )
        
        # 统计总数
        total = await db.projects.count_documents(query)
        
        # 获取用户置顶的项目ID列表
        user_id = ObjectId(current_user.get("user_id"))
        pinned_projects = await db.user_project_pins.find(
            {"user_id": user_id},
            {"project_id": 1}
        ).to_list(length=None)
        pinned_project_ids = {str(pin["project_id"]) for pin in pinned_projects}
        
        # 查询项目
        cursor = db.projects.find(query).sort("created_at", -1).skip(skip).limit(limit * 2)  # 多查询一些以便排序
        projects = await cursor.to_list(length=limit * 2)
        
        # 为每个项目添加 is_pinned 字段，并排序（置顶项目在前）
        for project in projects:
            project["is_pinned"] = str(project["_id"]) in pinned_project_ids
        
        # 排序：置顶项目在前，然后按创建时间倒序
        # 使用元组排序：(is_pinned的相反值, created_at的相反值)
        # is_pinned=True的项目会排在前面（因为not True = False < not False = True）
        def sort_key(p):
            is_pinned = p.get("is_pinned", False)
            created_at = p.get("created_at")
            if isinstance(created_at, datetime):
                # 返回 (not is_pinned, -timestamp) 用于排序
                # not True = False, not False = True，所以 False < True，置顶项目排在前面
                return (not is_pinned, -created_at.timestamp())
            return (not is_pinned, 0)
        
        projects.sort(key=sort_key)
        
        # 限制返回数量
        projects = projects[:limit]
        
        # 批量获取创建人信息
        creator_ids = []
        for project in projects:
            created_by = project.get("created_by")
            if created_by:
                creator_id_str = str(created_by)
                if creator_id_str and creator_id_str not in creator_ids:
                    creator_ids.append(creator_id_str)
        
        user_map = {}
        if creator_ids:
            try:
                creator_object_ids = []
                for uid in creator_ids:
                    try:
                        if ObjectId.is_valid(uid):
                            creator_object_ids.append(ObjectId(uid))
                    except Exception:
                        logger.warning(f"无效的创建人ID: {uid}")
                        continue
                
                if creator_object_ids:
                    users_cursor = db.users.find({"_id": {"$in": creator_object_ids}}, {"_id": 1, "name": 1, "username": 1})
                    users = await users_cursor.to_list(length=None)
                    user_map = {
                        str(user["_id"]): (user.get("name") or user.get("username") or "未知用户")
                        for user in users
                    }
                    logger.debug(f"成功获取 {len(user_map)} 个创建人信息")
            except Exception as e:
                logger.error(f"获取创建人信息失败: {e}", exc_info=True)
        
        # 转换为响应格式
        result = []
        for project in projects:
            # 获取任务统计
            task_count = await db.tasks.count_documents({"project_id": project["_id"]})
            # 从状态管理系统获取任务完成状态
            from app.utils.state_utils import get_completed_state_codes
            task_completed_statuses = await get_completed_state_codes(db, "task")
            completed_task_count = await db.tasks.count_documents({
                "project_id": project["_id"],
                "status": {"$in": task_completed_statuses} if task_completed_statuses else {"$in": []}
            })

            # 获取需求统计
            from app.utils.state_utils import get_completed_state_codes, get_state_code_by_name_keywords
            requirement_count = await db.requirements.count_documents({"project_id": project["_id"]})
            # 从状态管理系统获取完成状态列表
            completed_statuses = await get_completed_state_codes(db, "requirement")
            requirement_completed_count = await db.requirements.count_documents({
                "project_id": project["_id"],
                "status": {"$in": completed_statuses} if completed_statuses else {"$in": []}
            })
            # 从状态管理系统获取进行中状态
            in_progress_code = await get_state_code_by_name_keywords(db, "requirement", ["进行中", "in_progress", "testing"])
            active_statuses = [in_progress_code] if in_progress_code else []
            requirement_active_count = await db.requirements.count_documents({
                "project_id": project["_id"],
                "status": {"$in": active_statuses}
            })
            # 从状态管理系统获取待处理状态
            draft_code = await get_state_code_by_name_keywords(db, "requirement", ["草稿", "draft"])
            review_code = await get_state_code_by_name_keywords(db, "requirement", ["待审核", "review"])
            approved_code = await get_state_code_by_name_keywords(db, "requirement", ["已批准", "approved"])
            pending_statuses = [s for s in [draft_code, review_code, approved_code] if s]
            requirement_pending_count = await db.requirements.count_documents({
                "project_id": project["_id"],
                "status": {"$in": pending_statuses} if pending_statuses else {"$in": []}
            })

            # 获取关联模块统计
            test_case_count = await db.test_cases.count_documents({"project_id": project["_id"]})
            defect_count = await db.defects.count_documents({"project_id": project["_id"]})
            
            # 获取已解决缺陷数量
            resolved_defect_statuses = await get_completed_state_codes(db, "defect")
            resolved_defect_count = await db.defects.count_documents({
                "project_id": project["_id"],
                "status": {"$in": resolved_defect_statuses} if resolved_defect_statuses else {"$in": []}
            })
            
            # 获取已通过测试用例数量（测试用例通过状态）
            passed_test_case_statuses = await get_state_code_by_name_keywords(db, "test_case", ["通过", "passed", "成功"])
            passed_test_case_count = 0
            if passed_test_case_statuses:
                passed_test_case_count = await db.test_cases.count_documents({
                    "project_id": project["_id"],
                    "status": passed_test_case_statuses
                })
            else:
                # 如果没有找到通过状态，尝试从执行记录中统计
                # 这里简化处理，如果没有状态，暂时设为0
                passed_test_case_count = 0

            # 计算进度（综合多维度）
            progress = calculate_project_progress(
                requirement_count=requirement_count,
                completed_requirement_count=requirement_completed_count,
                task_count=task_count,
                completed_task_count=completed_task_count,
                defect_count=defect_count,
                resolved_defect_count=resolved_defect_count,
                test_case_count=test_case_count,
                passed_test_case_count=passed_test_case_count,
                project_status=project.get("status"),
                start_date=project.get("start_date"),
                end_date=project.get("end_date")
            )
            
            # 获取创建人名称
            created_by = project.get("created_by")
            created_by_id = str(created_by) if created_by else ""
            created_by_name = user_map.get(created_by_id, "未知用户") if created_by_id else "未知用户"
            
            # 确保 created_by_name 不为空
            if not created_by_name or created_by_name.strip() == "":
                created_by_name = "未知用户"
            
            project_response = self._format_project_response(
                project,
                task_count,
                completed_task_count,
                progress,
                extra_counts={
                    "requirement_count": requirement_count,
                    "requirement_completed_count": requirement_completed_count,
                    "requirement_active_count": requirement_active_count,
                    "requirement_pending_count": requirement_pending_count,
                    "test_case_count": test_case_count,
                    "defect_count": defect_count,
                    "created_by_name": created_by_name
                }
            )
            # 添加置顶状态
            project_response["is_pinned"] = project.get("is_pinned", False)
            result.append(project_response)
        
        return result, total

    async def get_project_by_id(
        self,
        db: AsyncIOMotorDatabase,
        project_id: str,
        current_user: dict
    ) -> Dict[str, Any]:
        """
        根据ID获取项目详情
        
        Args:
            db: 数据库连接
            project_id: 项目ID
            current_user: 当前用户信息
            
        Returns:
            项目详情字典
        """
        # 验证ObjectId格式
        try:
            object_id = ObjectId(project_id)
        except Exception:
            raise ValueError("Invalid project ID format")
        
        # 防止路由冲突
        if project_id == "stats":
            raise ValueError("Project not found")
        
        # 查找项目
        project = await db.projects.find_one({"_id": object_id})
        if not project:
            raise ValueError("Project not found")
        
        # 权限检查
        user_permissions = current_user.get("permissions", [])
        user_id = ObjectId(current_user.get("user_id"))
        
        has_access = (
            Permissions.PROJECT_READ_ALL in user_permissions or
            Permissions.ADMIN in user_permissions or
            project.get("is_public") or
            project.get("created_by") == user_id or
            any(member["user_id"] == user_id for member in project.get("members", []))
        )
        
        if not has_access:
            raise PermissionError("Not enough permissions to view this project")
        
        # 获取任务统计
        task_count = await db.tasks.count_documents({"project_id": project["_id"]})
        # 动态获取任务完成状态
        from app.utils.state_utils import get_completed_state_codes
        task_completed_statuses = await get_completed_state_codes(db, "task")
        completed_task_count = await db.tasks.count_documents({
            "project_id": project["_id"],
            "status": {"$in": task_completed_statuses} if task_completed_statuses else {"$in": []}
        })
 
        from app.utils.state_utils import get_completed_state_codes, get_state_code_by_name_keywords
        requirement_count = await db.requirements.count_documents({"project_id": project["_id"]})
        # 从状态管理系统获取完成状态列表
        completed_statuses = await get_completed_state_codes(db, "requirement")
        requirement_completed_count = await db.requirements.count_documents({
            "project_id": project["_id"],
            "status": {"$in": completed_statuses} if completed_statuses else {"$in": []}
        })
        # 从状态管理系统获取进行中状态
        in_progress_code = await get_state_code_by_name_keywords(db, "requirement", ["进行中", "in_progress", "testing"])
        active_statuses = [in_progress_code] if in_progress_code else []
        requirement_active_count = await db.requirements.count_documents({
            "project_id": project["_id"],
            "status": {"$in": active_statuses}
        })
        # 从状态管理系统获取待处理状态
        draft_code = await get_state_code_by_name_keywords(db, "requirement", ["草稿", "draft"])
        review_code = await get_state_code_by_name_keywords(db, "requirement", ["待审核", "review"])
        approved_code = await get_state_code_by_name_keywords(db, "requirement", ["已批准", "approved"])
        pending_statuses = [s for s in [draft_code, review_code, approved_code] if s]
        requirement_pending_count = await db.requirements.count_documents({
            "project_id": project["_id"],
            "status": {"$in": pending_statuses} if pending_statuses else {"$in": []}
        })

        # 获取关联模块统计
        test_case_count = await db.test_cases.count_documents({"project_id": project["_id"]})
        defect_count = await db.defects.count_documents({"project_id": project["_id"]})
        
        # 获取已解决缺陷数量
        resolved_defect_statuses = await get_completed_state_codes(db, "defect")
        resolved_defect_count = await db.defects.count_documents({
            "project_id": project["_id"],
            "status": {"$in": resolved_defect_statuses} if resolved_defect_statuses else {"$in": []}
        })
        
        # 获取已通过测试用例数量
        passed_test_case_statuses = await get_state_code_by_name_keywords(db, "test_case", ["通过", "passed", "成功"])
        passed_test_case_count = 0
        if passed_test_case_statuses:
            passed_test_case_count = await db.test_cases.count_documents({
                "project_id": project["_id"],
                "status": passed_test_case_statuses
            })

        # 计算进度（综合多维度）
        progress = calculate_project_progress(
            requirement_count=requirement_count,
            completed_requirement_count=requirement_completed_count,
            task_count=task_count,
            completed_task_count=completed_task_count,
            defect_count=defect_count,
            resolved_defect_count=resolved_defect_count,
            test_case_count=test_case_count,
            passed_test_case_count=passed_test_case_count,
            project_status=project.get("status"),
            start_date=project.get("start_date"),
            end_date=project.get("end_date")
        )

        # 获取用户置顶状态
        pinned_projects = await db.user_project_pins.find_one({
            "user_id": user_id,
            "project_id": object_id
        })
        project["is_pinned"] = pinned_projects is not None

        project_response = self._format_project_response(
            project,
            task_count,
            completed_task_count,
            progress,
            extra_counts={
                "requirement_count": requirement_count,
                "requirement_completed_count": requirement_completed_count,
                "requirement_active_count": requirement_active_count,
                "requirement_pending_count": requirement_pending_count,
                "test_case_count": test_case_count,
                "defect_count": defect_count
            }
        )
        # 添加置顶状态
        project_response["is_pinned"] = project.get("is_pinned", False)
        return project_response

    async def create_project(
        self,
        db: AsyncIOMotorDatabase,
        project_in: ProjectCreate,
        current_user: dict
    ) -> Dict[str, Any]:
        """
        创建项目
        
        Args:
            db: 数据库连接
            project_in: 项目创建数据
            current_user: 当前用户信息
            
        Returns:
            创建的项目详情
        """
        # 检查权限
        user_permissions = current_user.get("permissions", [])
        if Permissions.PROJECT_CREATE not in user_permissions and Permissions.ADMIN not in user_permissions:
            raise PermissionError("Not enough permissions to create project")
        
        # 检查项目名称是否已存在
        existing_project = await db.projects.find_one({"name": project_in.name})
        if existing_project:
            raise ValueError("Project name already exists")
        
        # 验证优先级
        if project_in.priority:
            is_valid = await self.validate_priority(db, project_in.priority)
            if not is_valid:
                raise ValueError(f"Invalid priority value: {project_in.priority}")
        
        # 创建项目文档
        project_doc = project_in.model_dump(exclude={"owner_id"})

        # 将 date 对象转换为 datetime 对象
        if project_doc.get("start_date"):
            project_doc["start_date"] = datetime.combine(project_doc["start_date"], datetime.min.time())
        if project_doc.get("end_date"):
            project_doc["end_date"] = datetime.combine(project_doc["end_date"], datetime.min.time())

        project_doc.update({
            "created_by": ObjectId(current_user.get("user_id")),
            "created_at": datetime.utcnow(),
            "updated_at": datetime.utcnow(),
            "is_archived": False
        })
        
        # 转换成员中的user_id为ObjectId
        if "members" in project_doc:
            for member in project_doc["members"]:
                member["user_id"] = ObjectId(member["user_id"])
        
        # 处理项目负责人和创建者
        creator_id = ObjectId(current_user.get("user_id"))
        owner_id = None

        if project_in.owner_id:
            try:
                owner_id = ObjectId(project_in.owner_id)
                owner_user = await db.users.find_one({"_id": owner_id})
                if not owner_user:
                    raise ValueError("Designated owner user not found")
            except Exception:
                raise ValueError("Invalid owner_id format")
        else:
            owner_id = creator_id

        if "members" not in project_doc:
            project_doc["members"] = []
        
        member_ids = {member["user_id"] for member in project_doc["members"]}

        # 添加负责人（如果不存在）
        if owner_id not in member_ids:
            # 获取Owner角色ID
            owner_role_doc = await db.roles.find_one({"code": "owner"})
            owner_role_id = str(owner_role_doc["_id"]) if owner_role_doc else "owner"
            
            project_doc["members"].append({
                "user_id": owner_id,
                "roles": [owner_role_id],
                "joined_at": datetime.utcnow(),
                "permissions": ["project:read", "project:update", "project:delete", "project:manage_members"]
            })
            member_ids.add(owner_id)

        # 插入数据库
        result = await db.projects.insert_one(project_doc)
        
        # 获取创建的项目
        created_project = await db.projects.find_one({"_id": result.inserted_id})
        if not created_project:
            raise RuntimeError("Failed to retrieve created project")
        
        return self._format_project_response(
            created_project,
            0,
            0,
            0.0,
            extra_counts={
                "requirement_count": 0,
                "requirement_completed_count": 0,
                "requirement_active_count": 0,
                "requirement_pending_count": 0,
                "test_case_count": 0,
                "defect_count": 0
            }
        )

    async def update_project(
        self,
        db: AsyncIOMotorDatabase,
        project_id: str,
        project_in: ProjectUpdate,
        current_user: dict
    ) -> Dict[str, Any]:
        """
        更新项目
        
        Args:
            db: 数据库连接
            project_id: 项目ID
            project_in: 项目更新数据
            current_user: 当前用户信息
            
        Returns:
            更新后的项目详情
        """
        # 验证ObjectId格式
        try:
            object_id = ObjectId(project_id)
        except Exception:
            raise ValueError("Invalid project ID format")
        
        # 检查项目是否存在
        project = await db.projects.find_one({"_id": object_id})
        if not project:
            raise ValueError("Project not found")
        
        # 权限检查
        user_permissions = current_user.get("permissions", [])
        user_id = ObjectId(current_user.get("user_id"))
        
        # 检查是否有权限更新项目
        has_permission = (
            Permissions.PROJECT_UPDATE_ALL in user_permissions or
            Permissions.ADMIN in user_permissions or
            project["created_by"] == user_id or
            any(
                member["user_id"] == user_id and "project:update" in member.get("permissions", [])
                for member in project.get("members", [])
            )
        )
        
        if not has_permission:
            raise PermissionError("Not enough permissions to update this project")
        
        # 如果更新项目名称，检查是否已存在
        update_data = project_in.model_dump(exclude_unset=True)
        if "name" in update_data and update_data["name"] != project["name"]:
            existing_project = await db.projects.find_one({"name": update_data["name"]})
            if existing_project:
                raise ValueError("Project name already exists")
        
        # 验证优先级
        if "priority" in update_data and update_data["priority"]:
            is_valid = await self.validate_priority(db, update_data["priority"])
            if not is_valid:
                raise ValueError(f"Invalid priority value: {update_data['priority']}")
        
        # 处理负责人更新
        new_owner_id_str = update_data.pop("owner_id", None)
        if new_owner_id_str is not None:
            to_object_id = getattr(self, "_to_object_id", None)

            def _convert_object_id(value: Any, field_name: str = "ID") -> Optional[ObjectId]:
                if callable(to_object_id):
                    return to_object_id(value, field_name)
                if value is None:
                    return None
                if isinstance(value, ObjectId):
                    return value
                if isinstance(value, str) and ObjectId.is_valid(value):
                    return ObjectId(value)
                raise ValueError(f"Invalid {field_name} format")

            new_owner_object_id = _convert_object_id(new_owner_id_str, "负责人ID")
            if not new_owner_object_id:
                raise ValueError("Invalid owner_id format")
            
            new_owner_user = await db.users.find_one({"_id": new_owner_object_id})
            if not new_owner_user:
                raise ValueError("Designated owner user not found")

            members = project.get("members", [])
            updated_members = []
            new_owner_found = False

            for member in members:
                member_user_id = member.get("user_id")
                # 兼容字符串与ObjectId
                if isinstance(member_user_id, ObjectId):
                    member_user_id_obj = member_user_id
                else:
                    try:
                        member_user_id_obj = _convert_object_id(member_user_id)
                    except ValueError:
                        member_user_id_obj = None
                member_user_id_str = str(member_user_id_obj) if member_user_id_obj else str(member_user_id)

                updated_member = dict(member)

                if member_user_id_obj is None and member_user_id:
                    # 将可转换的字符串转为ObjectId
                    try:
                        maybe_obj = _convert_object_id(member_user_id)
                    except ValueError:
                        maybe_obj = None
                    if maybe_obj is not None:
                        member_user_id_obj = maybe_obj
                        member_user_id_str = str(maybe_obj)
                        updated_member["user_id"] = maybe_obj

                if member_user_id_str == str(new_owner_object_id):
                    new_owner_found = True
                    updated_member["user_id"] = new_owner_object_id
                    
                    # 获取Owner角色ID
                    owner_role_doc = await db.roles.find_one({"code": "owner"})
                    owner_role_id = str(owner_role_doc["_id"]) if owner_role_doc else "owner"
                    updated_member["roles"] = [owner_role_id]
                    
                    updated_member["permissions"] = [
                        "project:read",
                        "project:update",
                        "project:delete",
                        "project:manage_members"
                    ]
                elif "owner" in (member.get("roles", []) if isinstance(member.get("roles"), list) else [member.get("role")]):
                    # 原负责人降级为项目管理员
                    # 获取Manager角色ID
                    manager_role_doc = await db.roles.find_one({"code": "project_manager"})
                    manager_role_id = str(manager_role_doc["_id"]) if manager_role_doc else "project_manager"
                    updated_member["roles"] = [manager_role_id]
                    
                    perms = set(member.get("permissions", []))
                    perms.update({"project:read", "project:update", "project:manage_members"})
                    perms.discard("project:delete")
                    updated_member["permissions"] = list(perms)
                    update_data["project_manager"] = str(member_user_id_obj or member_user_id)

                updated_members.append(updated_member)

            if not new_owner_found:
                # 获取Owner角色ID
                owner_role_doc = await db.roles.find_one({"code": "owner"})
                owner_role_id = str(owner_role_doc["_id"]) if owner_role_doc else "owner"
                
                updated_members.append({
                    "user_id": new_owner_object_id,
                    "roles": [owner_role_id],
                    "joined_at": datetime.utcnow(),
                    "permissions": [
                        "project:read",
                        "project:update",
                        "project:delete",
                        "project:manage_members"
                    ]
                })

            update_data["members"] = updated_members
            update_data.pop("project_manager", None)

        # 更新项目
        if update_data:
            # 将 date 对象转换为 datetime 对象，避免 MongoDB 序列化错误
            if "start_date" in update_data:
                start_value = update_data["start_date"]
                if isinstance(start_value, date) and not isinstance(start_value, datetime):
                    update_data["start_date"] = datetime.combine(start_value, datetime.min.time())
            if "end_date" in update_data:
                end_value = update_data["end_date"]
                if isinstance(end_value, date) and not isinstance(end_value, datetime):
                    update_data["end_date"] = datetime.combine(end_value, datetime.min.time())

            update_data["updated_at"] = datetime.utcnow()
            await db.projects.update_one(
                {"_id": object_id},
                {"$set": update_data}
            )
        
        # 获取更新后的项目
        updated_project = await db.projects.find_one({"_id": object_id})
        
        # 获取任务统计
        task_count = await db.tasks.count_documents({"project_id": updated_project["_id"]})
        # 动态获取任务完成状态
        from app.utils.state_utils import get_completed_state_codes, get_state_code_by_name_keywords
        task_completed_statuses = await get_completed_state_codes(db, "task")
        completed_task_count = await db.tasks.count_documents({
            "project_id": updated_project["_id"],
            "status": {"$in": task_completed_statuses} if task_completed_statuses else {"$in": []}
        })
 
        requirement_count = await db.requirements.count_documents({"project_id": updated_project["_id"]})
        # 动态获取需求完成状态
        requirement_completed_statuses = await get_completed_state_codes(db, "requirement")
        requirement_completed_count = await db.requirements.count_documents({
            "project_id": updated_project["_id"],
            "status": {"$in": requirement_completed_statuses} if requirement_completed_statuses else {"$in": []}
        })
        # 动态获取需求进行中状态
        requirement_active_code = await get_state_code_by_name_keywords(db, "requirement", ["进行中", "in_progress", "testing"])
        requirement_active_count = await db.requirements.count_documents({
            "project_id": updated_project["_id"],
            "status": requirement_active_code
        }) if requirement_active_code else 0
        # 动态获取需求草稿状态
        requirement_draft_code = await get_state_code_by_name_keywords(db, "requirement", ["草稿", "draft"])
        requirement_pending_count = await db.requirements.count_documents({
            "project_id": updated_project["_id"],
            "status": requirement_draft_code
        }) if requirement_draft_code else 0

        # 获取关联模块统计
        test_case_count = await db.test_cases.count_documents({"project_id": updated_project["_id"]})
        defect_count = await db.defects.count_documents({"project_id": updated_project["_id"]})
        
        # 获取已解决缺陷数量
        resolved_defect_statuses = await get_completed_state_codes(db, "defect")
        resolved_defect_count = await db.defects.count_documents({
            "project_id": updated_project["_id"],
            "status": {"$in": resolved_defect_statuses} if resolved_defect_statuses else {"$in": []}
        })
        
        # 获取已通过测试用例数量
        passed_test_case_statuses = await get_state_code_by_name_keywords(db, "test_case", ["通过", "passed", "成功"])
        passed_test_case_count = 0
        if passed_test_case_statuses:
            passed_test_case_count = await db.test_cases.count_documents({
                "project_id": updated_project["_id"],
                "status": passed_test_case_statuses
            })

        # 计算进度（综合多维度）
        progress = calculate_project_progress(
            requirement_count=requirement_count,
            completed_requirement_count=requirement_completed_count,
            task_count=task_count,
            completed_task_count=completed_task_count,
            defect_count=defect_count,
            resolved_defect_count=resolved_defect_count,
            test_case_count=test_case_count,
            passed_test_case_count=passed_test_case_count,
            project_status=updated_project.get("status"),
            start_date=updated_project.get("start_date"),
            end_date=updated_project.get("end_date")
        )

        return self._format_project_response(
            updated_project,
            task_count,
            completed_task_count,
            progress,
            extra_counts={
                "requirement_count": requirement_count,
                "requirement_completed_count": requirement_completed_count,
                "requirement_active_count": requirement_active_count,
                "requirement_pending_count": requirement_pending_count,
                "test_case_count": test_case_count,
                "defect_count": defect_count
            }
        )

    async def delete_project(
        self,
        db: AsyncIOMotorDatabase,
        project_id: str,
        current_user: dict
    ) -> bool:
        """
        删除项目（软删除：归档）
        
        Args:
            db: 数据库连接
            project_id: 项目ID
            current_user: 当前用户信息
            
        Returns:
            是否成功
        """
        from app.utils.state_utils import get_state_code_by_name_keywords
        
        # 验证ObjectId格式
        try:
            object_id = ObjectId(project_id)
        except Exception:
            raise ValueError("Invalid project ID format")
        
        # 检查项目是否存在
        project = await db.projects.find_one({"_id": object_id})
        if not project:
            raise ValueError("Project not found")
        
        # 权限检查
        user_permissions = current_user.get("permissions", [])
        user_id = ObjectId(current_user.get("user_id"))
        
        has_permission = (
            Permissions.PROJECT_DELETE in user_permissions or
            Permissions.ADMIN in user_permissions or
            project["created_by"] == user_id or
            any(
                member["user_id"] == user_id and "project:delete" in member.get("permissions", [])
                for member in project.get("members", [])
            )
        )
        
        if not has_permission:
            raise PermissionError("Not enough permissions to delete this project")
        
        # 获取归档状态代码
        archived_status_code = await get_state_code_by_name_keywords(db, "project", ["已归档", "归档", "archived"]) or "archived"
        
        # 软删除：归档项目
        await db.projects.update_one(
            {"_id": object_id},
            {
                "$set": {
                    "is_archived": True,
                    "status": archived_status_code,
                    "updated_at": datetime.utcnow()
                }
            }
        )
        
        return True

    async def archive_project(
        self,
        db: AsyncIOMotorDatabase,
        project_id: str,
        current_user: dict
    ) -> Dict[str, Any]:
        """
        归档项目
        
        Args:
            db: 数据库连接
            project_id: 项目ID
            current_user: 当前用户信息
            
        Returns:
            归档后的项目数据
        """
        from app.utils.state_utils import get_state_code_by_name_keywords
        
        # 验证ObjectId格式
        try:
            object_id = ObjectId(project_id)
        except Exception:
            raise ValueError("Invalid project ID format")
        
        # 检查项目是否存在
        project = await db.projects.find_one({"_id": object_id})
        if not project:
            raise ValueError("Project not found")
        
        # 检查项目是否已经归档
        if project.get("is_archived", False):
            raise ValueError("Project is already archived")
        
        # 权限检查
        user_permissions = current_user.get("permissions", [])
        user_id = ObjectId(current_user.get("user_id"))
        
        has_permission = (
            Permissions.PROJECT_UPDATE in user_permissions or
            Permissions.PROJECT_DELETE in user_permissions or
            Permissions.ADMIN in user_permissions or
            project["created_by"] == user_id or
            any(
                member["user_id"] == user_id and (
                    "project:update" in member.get("permissions", []) or
                    "project:delete" in member.get("permissions", [])
                )
                for member in project.get("members", [])
            )
        )
        
        if not has_permission:
            raise PermissionError("Not enough permissions to archive this project")
        
        # 获取归档状态代码
        archived_status_code = await get_state_code_by_name_keywords(
            db, "project", ["已归档", "归档", "archived"]
        ) or "archived"
        
        # 归档项目
        now = datetime.utcnow()
        await db.projects.update_one(
            {"_id": object_id},
            {
                "$set": {
                    "is_archived": True,
                    "status": archived_status_code,
                    "archived_at": now,
                    "updated_at": now
                }
            }
        )
        
        # 清除仪表盘缓存
        self.invalidate_dashboard_cache()
        
        # 返回更新后的项目数据
        updated_project = await db.projects.find_one({"_id": object_id})
        return self._format_project_response(updated_project)

    async def unarchive_project(
        self,
        db: AsyncIOMotorDatabase,
        project_id: str,
        current_user: dict
    ) -> Dict[str, Any]:
        """
        取消归档项目（恢复项目）
        
        Args:
            db: 数据库连接
            project_id: 项目ID
            current_user: 当前用户信息
            
        Returns:
            恢复后的项目数据
        """
        from app.utils.state_utils import get_state_code_by_name_keywords, get_state_options
        
        # 验证ObjectId格式
        try:
            object_id = ObjectId(project_id)
        except Exception:
            raise ValueError("Invalid project ID format")
        
        # 检查项目是否存在
        project = await db.projects.find_one({"_id": object_id})
        if not project:
            raise ValueError("Project not found")
        
        # 检查项目是否已归档
        if not project.get("is_archived", False):
            raise ValueError("Project is not archived")
        
        # 权限检查
        user_permissions = current_user.get("permissions", [])
        user_id = ObjectId(current_user.get("user_id"))
        
        has_permission = (
            Permissions.PROJECT_UPDATE in user_permissions or
            Permissions.PROJECT_DELETE in user_permissions or
            Permissions.ADMIN in user_permissions or
            project["created_by"] == user_id or
            any(
                member["user_id"] == user_id and (
                    "project:update" in member.get("permissions", []) or
                    "project:delete" in member.get("permissions", [])
                )
                for member in project.get("members", [])
            )
        )
        
        if not has_permission:
            raise PermissionError("Not enough permissions to unarchive this project")
        
        # 从状态管理系统获取默认状态
        from app.utils.state_utils import get_default_state_code
        default_status = await get_default_state_code(db, "project")
        if not default_status:
            # 如果无法获取默认状态，尝试查找"规划中"或"进行中"状态
            from app.utils.state_utils import get_state_code_by_name_keywords
            default_status = await get_state_code_by_name_keywords(db, "project", ["规划中", "planning"])
            if not default_status:
                default_status = await get_state_code_by_name_keywords(db, "project", ["进行中", "active"])
            if not default_status:
                raise ValueError("无法获取项目的默认状态，请检查状态管理系统配置")
        
        # 取消归档项目
        now = datetime.utcnow()
        await db.projects.update_one(
            {"_id": object_id},
            {
                "$set": {
                    "is_archived": False,
                    "status": default_status,
                    "archived_at": None,
                    "updated_at": now
                },
                "$unset": {
                    "archived_at": ""
                }
            }
        )
        
        # 清除仪表盘缓存
        self.invalidate_dashboard_cache()
        
        # 返回更新后的项目数据
        updated_project = await db.projects.find_one({"_id": object_id})
        return self._format_project_response(updated_project)

    async def get_dashboard_stats(
        self,
        db: AsyncIOMotorDatabase,
        current_user: dict,
        start_date: Optional[str] = None,
        end_date: Optional[str] = None,
        period: str = "month",
        force_refresh: bool = False
    ) -> Dict[str, Any]:
        """
        获取项目仪表盘统计数据（带缓存）
        
        Args:
            db: 数据库连接
            current_user: 当前用户信息
            start_date: 开始日期（YYYY-MM-DD）
            end_date: 结束日期（YYYY-MM-DD）
            period: 趋势周期：week/month/quarter
            force_refresh: 是否强制刷新缓存
            
        Returns:
            仪表盘数据字典
        """
        # 验证参数
        if period not in ["week", "month", "quarter"]:
            period = "month"
            logger.warning(f"Invalid period '{period}', using default 'month'")
        
        # 验证日期格式
        if start_date:
            try:
                datetime.strptime(start_date, "%Y-%m-%d")
            except ValueError:
                logger.warning(f"Invalid start_date format: {start_date}")
                start_date = None
        
        if end_date:
            try:
                datetime.strptime(end_date, "%Y-%m-%d")
            except ValueError:
                logger.warning(f"Invalid end_date format: {end_date}")
                end_date = None
        
        # 生成缓存键
        user_id = current_user.get("user_id", "")
        cache_key = self._generate_dashboard_cache_key(
            user_id, start_date, end_date, period
        )
        
        # 检查缓存
        async with self._dashboard_cache_lock:
            if not force_refresh and self._is_dashboard_cache_valid(cache_key):
                logger.debug(f"从缓存获取仪表盘数据: {cache_key}")
                return self._dashboard_cache[cache_key].copy()
        
        try:
            # 构建查询条件（基于用户权限）
            query = {}
            user_permissions = current_user.get("permissions", [])
            user_id_obj = ObjectId(user_id)
            
            if Permissions.PROJECT_READ_ALL not in user_permissions and Permissions.ADMIN not in user_permissions:
                query["$or"] = [
                    {"is_public": True},
                    {"created_by": user_id_obj},
                    {"members.user_id": user_id_obj}
                ]
            
            # 排除归档项目（仪表盘统计不包含归档项目）
            query["is_archived"] = {"$ne": True}
            
            # 并行获取所有统计数据
            progress_distribution, status_distribution, trend_result, team_efficiency = await asyncio.gather(
                self._get_progress_distribution(db, query),
                self._get_status_distribution(db, query),
                self._get_creation_trend(db, query, start_date, end_date, period),
                self._get_team_efficiency(db, query),
                return_exceptions=True
            )
            
            # 处理异常并统一转换为默认值
            def safe_result(value: Any, default: Any) -> Any:
                """安全获取结果，异常时返回默认值"""
                if isinstance(value, Exception):
                    logger.error(f"获取数据失败: {value}")
                    return default
                return value
            
            progress_distribution = safe_result(progress_distribution, [])
            status_distribution = safe_result(status_distribution, [])
            
            if isinstance(trend_result, Exception):
                logger.error(f"获取趋势数据失败: {trend_result}")
                trend_data, trend_labels = [], []
            else:
                trend_data, trend_labels = trend_result
            
            team_efficiency = safe_result(team_efficiency, [])
            
            result = {
                "progress_distribution": progress_distribution,
                "status_distribution": status_distribution,
                "trend_data": {
                    "labels": trend_labels,
                    "created": [item["created"] for item in trend_data] if trend_data else [],
                    "completed": [item["completed"] for item in trend_data] if trend_data else []
                },
                "team_efficiency": team_efficiency
            }
            
            # 更新缓存
            async with self._dashboard_cache_lock:
                self._dashboard_cache[cache_key] = result
                self._dashboard_cache_timestamps[cache_key] = datetime.now(timezone.utc)
            
            return result
            
        except Exception as e:
            logger.error(f"获取仪表盘统计数据失败: {str(e)}", exc_info=True)
            # 返回空数据而不是抛出异常
            return self._get_empty_dashboard_data()
    
    def _generate_dashboard_cache_key(
        self,
        user_id: str,
        start_date: Optional[str],
        end_date: Optional[str],
        period: str
    ) -> str:
        """生成仪表盘缓存键"""
        key_data = {
            "user_id": user_id,
            "start_date": start_date or "",
            "end_date": end_date or "",
            "period": period
        }
        key_str = json.dumps(key_data, sort_keys=True)
        return f"dashboard_stats_{hashlib.md5(key_str.encode()).hexdigest()}"
    
    def _get_empty_dashboard_data(self) -> Dict[str, Any]:
        """返回空的仪表盘数据结构"""
        return {
            "progress_distribution": [],
            "status_distribution": [],
            "trend_data": {
                "labels": [],
                "created": [],
                "completed": []
            },
            "team_efficiency": []
        }
    
    def _is_dashboard_cache_valid(self, cache_key: str) -> bool:
        """检查仪表盘缓存是否有效"""
        if cache_key not in self._dashboard_cache:
            return False
        
        timestamp = self._dashboard_cache_timestamps.get(cache_key)
        if not timestamp:
            return False
        
        elapsed = datetime.now(timezone.utc) - timestamp
        return elapsed < self._dashboard_cache_ttl
    
    async def invalidate_dashboard_cache(self, user_id: Optional[str] = None):
        """
        清除仪表盘缓存
        
        Args:
            user_id: 用户ID，如果提供则只清除该用户的缓存，否则清除所有缓存
        """
        async with self._dashboard_cache_lock:
            if user_id:
                # 清除特定用户的缓存
                keys_to_remove = [
                    key for key in self._dashboard_cache.keys()
                    if user_id in key
                ]
                for key in keys_to_remove:
                    self._dashboard_cache.pop(key, None)
                    self._dashboard_cache_timestamps.pop(key, None)
            else:
                # 清除所有缓存
                self._dashboard_cache.clear()
                self._dashboard_cache_timestamps.clear()
            logger.debug(f"清除仪表盘缓存: user_id={user_id}")

    async def _get_progress_distribution(
        self,
        db: AsyncIOMotorDatabase,
        query: Dict[str, Any]
    ) -> List[Dict[str, Any]]:
        """获取进度分布（优化版本）"""
        progress_ranges = [
            {"min": 0, "max": 25, "name": "0-25%"},
            {"min": 26, "max": 50, "name": "26-50%"},
            {"min": 51, "max": 75, "name": "51-75%"},
            {"min": 76, "max": 100, "name": "76-100%"}
        ]
        
        try:
            # 获取所有项目ID（只查询_id字段，减少数据传输）
            project_ids_cursor = db.projects.find(query, {"_id": 1})
            project_ids_list = [p["_id"] async for p in project_ids_cursor]
            
            if not project_ids_list:
                # 如果无项目，返回空的进度分布
                return [
                    {"value": 0, "name": range_item["name"]}
                    for range_item in progress_ranges
                ]
            
            # 动态获取任务和需求的完成状态
            from app.utils.state_utils import get_completed_state_codes
            task_completed_statuses = await get_completed_state_codes(db, "task")
            requirement_completed_statuses = await get_completed_state_codes(db, "requirement")
            
            # 并行获取任务和需求统计
            task_stats_pipeline = [
                {"$match": {"project_id": {"$in": project_ids_list}}},
                {"$group": {
                    "_id": "$project_id",
                    "total_tasks": {"$sum": 1},
                    "completed_tasks": {
                        "$sum": {
                            "$cond": [
                                {"$in": ["$status", task_completed_statuses] if task_completed_statuses else []},
                                1,
                                0
                            ]
                        }
                    }
                }}
            ]

            requirement_stats_pipeline = [
                {"$match": {"project_id": {"$in": project_ids_list}}},
                {"$group": {
                    "_id": "$project_id",
                    "total_requirements": {"$sum": 1},
                    "completed_requirements": {
                        "$sum": {
                            "$cond": [
                                {"$in": ["$status", requirement_completed_statuses] if requirement_completed_statuses else []},
                                1,
                                0
                            ]
                        }
                    }
                }}
            ]

            # 并行执行聚合查询
            task_stats, requirement_stats = await asyncio.gather(
                db.tasks.aggregate(task_stats_pipeline).to_list(length=None),
                db.requirements.aggregate(requirement_stats_pipeline).to_list(length=None)
            )

            # 构建项目ID到统计的映射（使用字典提高查找效率）
            task_stats_map = {
                stat["_id"]: (stat["total_tasks"], stat["completed_tasks"])
                for stat in task_stats
            }

            requirement_stats_map = {
                stat["_id"]: (stat["total_requirements"], stat["completed_requirements"])
                for stat in requirement_stats
            }

            # 统计进度分布
            progress_counts = [0, 0, 0, 0]  # 对应4个进度区间

            for project_id in project_ids_list:
                total_tasks, completed_tasks = task_stats_map.get(project_id, (0, 0))
                total_requirements, completed_requirements = requirement_stats_map.get(project_id, (0, 0))
                
                try:
                    # 获取项目信息以获取状态和日期
                    project = await db.projects.find_one({"_id": project_id})
                    project_status = project.get("status") if project else None
                    start_date = project.get("start_date") if project else None
                    end_date = project.get("end_date") if project else None
                    
                    # 获取缺陷和测试用例统计（简化处理，使用0作为默认值）
                    progress = calculate_project_progress(
                        requirement_count=total_requirements,
                        completed_requirement_count=completed_requirements,
                        task_count=total_tasks,
                        completed_task_count=completed_tasks,
                        defect_count=0,  # 统计摘要中暂不包含缺陷和测试用例
                        resolved_defect_count=0,
                        test_case_count=0,
                        passed_test_case_count=0,
                        project_status=project_status,
                        start_date=start_date,
                        end_date=end_date
                    )
                except Exception as e:
                    logger.warning(f"计算项目进度失败 (project_id={project_id}): {e}")
                    progress = 0

                # 统计到对应区间
                if 0 <= progress <= 25:
                    progress_counts[0] += 1
                elif 26 <= progress <= 50:
                    progress_counts[1] += 1
                elif 51 <= progress <= 75:
                    progress_counts[2] += 1
                elif 76 <= progress <= 100:
                    progress_counts[3] += 1
            
            return [
                {"value": progress_counts[i], "name": range_item["name"]}
                for i, range_item in enumerate(progress_ranges)
            ]
            
        except Exception as e:
            logger.error(f"获取进度分布失败: {str(e)}", exc_info=True)
            # 返回空数据
            return [
                {"value": 0, "name": range_item["name"]}
                for range_item in progress_ranges
            ]

    async def _get_status_distribution(
        self,
        db: AsyncIOMotorDatabase,
        query: Dict[str, Any]
    ) -> List[Dict[str, Any]]:
        """获取状态分布"""
        pipeline = [
            {"$match": query},
            {"$group": {"_id": "$status", "count": {"$sum": 1}}}
        ]
        status_stats = await db.projects.aggregate(pipeline).to_list(length=None)
        
        # 获取所有状态以填充缺失的状态
        all_states = await db.states.find({"module_type": "project", "is_enabled": True}).to_list(length=None)
        status_map = {}
        for stat in status_stats:
            status_map[stat["_id"]] = stat["count"]
        
        status_distribution = []
        for state in all_states:
            status_distribution.append({
                "value": status_map.get(state["code"], 0),
                "name": state["name"],
                "code": state["code"]
            })
        
        return status_distribution

    async def _get_creation_trend(
        self,
        db: AsyncIOMotorDatabase,
        query: Dict[str, Any],
        start_date: Optional[str],
        end_date: Optional[str],
        period: str
    ) -> tuple[List[Dict[str, Any]], List[str]]:
        """获取创建趋势（使用聚合避免加载全部项目）"""

        def _normalize_period_start(dt: datetime, period_key: str) -> datetime:
            dt = dt.replace(hour=0, minute=0, second=0, microsecond=0)
            if period_key == "week":
                return dt - timedelta(days=dt.weekday())
            if period_key == "month":
                return dt.replace(day=1)
            if period_key == "quarter":
                quarter_start_month = ((dt.month - 1) // 3) * 3 + 1
                return dt.replace(month=quarter_start_month, day=1)
            return dt

        def _next_period_start(dt: datetime, period_key: str) -> datetime:
            if period_key == "week":
                return dt + timedelta(weeks=1)
            if period_key == "month":
                year = dt.year + (dt.month // 12)
                month = dt.month % 12 + 1
                return dt.replace(year=year, month=month, day=1)
            if period_key == "quarter":
                next_month = dt.month + 3
                year = dt.year + ((next_month - 1) // 12)
                month = ((next_month - 1) % 12) + 1
                return dt.replace(year=year, month=month, day=1)
            return dt

        def _format_period_key(dt: datetime, period_key: str) -> Tuple[str, str]:
            if period_key == "week":
                iso_year, iso_week, _ = dt.isocalendar()
                return f"{iso_year}-W{iso_week:02d}", f"{dt.month}/{dt.day}"
            if period_key == "quarter":
                quarter = (dt.month - 1) // 3 + 1
                return f"{dt.year}-Q{quarter}", f"{dt.year}Q{quarter}"
            # month
            return f"{dt.year}-{dt.month:02d}", f"{dt.year}-{dt.month:02d}"

        def _period_unit(p: str) -> str:
            if p == "week":
                return "week"
            if p == "quarter":
                return "quarter"
            return "month"

        now = datetime.utcnow()
        end_dt = now
        if end_date:
            try:
                end_dt = datetime.strptime(end_date, "%Y-%m-%d")
            except ValueError:
                end_dt = now

        start_dt = end_dt - timedelta(days=365)
        if start_date:
            try:
                start_dt = datetime.strptime(start_date, "%Y-%m-%d")
            except ValueError:
                pass

        if start_dt > end_dt:
            start_dt, end_dt = end_dt, start_dt

        period_unit = _period_unit(period)
        start_dt = _normalize_period_start(start_dt, period_unit)
        end_dt = _normalize_period_start(end_dt, period_unit)
        end_dt = end_dt.replace(hour=23, minute=59, second=59, microsecond=999999)

        created_match = {**query, "created_at": {"$gte": start_dt, "$lte": end_dt}}
        created_pipeline = [
            {"$match": created_match},
            {
                "$group": {
                    "_id": {
                        "period": {
                            "$dateTrunc": {
                                "date": "$created_at",
                                "unit": period_unit
                            }
                        }
                    },
                    "count": {"$sum": 1}
                }
            },
            {"$sort": {"_id.period": 1}}
        ]
        created_counts = await db.projects.aggregate(created_pipeline).to_list(length=None)
        created_map: Dict[str, int] = {}
        for item in created_counts:
            period_start = item["_id"]["period"]
            key, _ = _format_period_key(period_start, period_unit)
            created_map[key] = item["count"]

        completed_map: Dict[str, int] = {}
        completed_status_code = await get_completed_state_code(db, "project")
        if completed_status_code:
            completed_match = {
                **query,
                "status": completed_status_code,
                "updated_at": {"$gte": start_dt, "$lte": end_dt}
            }
            completed_pipeline = [
                {"$match": completed_match},
                {
                    "$group": {
                        "_id": {
                            "period": {
                                "$dateTrunc": {
                                    "date": "$updated_at",
                                    "unit": period_unit
                                }
                            }
                        },
                        "count": {"$sum": 1}
                    }
                },
                {"$sort": {"_id.period": 1}}
            ]
            completed_counts = await db.projects.aggregate(completed_pipeline).to_list(length=None)
            for item in completed_counts:
                period_start = item["_id"]["period"]
                key, _ = _format_period_key(period_start, period_unit)
                completed_map[key] = item["count"]

        trend_data: List[Dict[str, Any]] = []
        trend_labels: List[str] = []

        current = start_dt
        safe_guard = 0
        while current <= end_dt and safe_guard < 500:
            key, label = _format_period_key(current, period_unit)
            trend_labels.append(label)
            trend_data.append({
                "period": label,
                "created": created_map.get(key, 0),
                "completed": completed_map.get(key, 0)
            })
            current = _next_period_start(current, period_unit)
            safe_guard += 1

        return trend_data, trend_labels

    async def _get_team_efficiency(
        self,
        db: AsyncIOMotorDatabase,
        query: Dict[str, Any]
    ) -> List[Dict[str, Any]]:
        """获取团队效率排行"""
        team_efficiency = []
        
        # 获取已完成状态代码
        completed_status_code = await get_completed_state_code(db, "project")
        
        # 获取所有项目成员及其完成的项目数
        pipeline = [
            {"$match": query},
            {"$unwind": "$members"},
            {"$group": {
                "_id": "$members.user_id",
                "total_projects": {"$sum": 1},
                "completed_projects": {
                    "$sum": {"$cond": [{"$eq": ["$status", completed_status_code]}, 1, 0]}
                }
            }},
            {"$sort": {"completed_projects": -1}},
            {"$limit": 10}
        ]
        
        member_stats = await db.projects.aggregate(pipeline).to_list(length=None)
        
        if member_stats:
            # 批量获取用户信息
            member_ids = [stat["_id"] for stat in member_stats]
            users = await db.users.find({"_id": {"$in": member_ids}}).to_list(length=None)
            user_map = {str(user["_id"]): user for user in users}
            
            for stat in member_stats:
                member_id = stat["_id"]
                # 获取成员信息
                user = user_map.get(str(member_id))
                if user:
                    team_efficiency.append({
                        "user_id": str(member_id),
                        "name": user.get("name") or user.get("username") or "未知用户",
                        "completed_count": stat["completed_projects"],
                        "total_count": stat["total_projects"]
                    })
        
        return team_efficiency

    async def get_project_activities(
        self,
        db: AsyncIOMotorDatabase,
        project_id: str,
        current_user: dict,
        page: int = 1,
        size: int = 20
    ) -> Dict[str, Any]:
        """获取项目相关活动日志"""
        try:
            object_id = ObjectId(project_id)
        except Exception:
            raise ValueError("Invalid project ID format")

        project = await db.projects.find_one({"_id": object_id})
        if not project:
            raise ValueError("Project not found")

        user_permissions = current_user.get("permissions", [])
        user_id = ObjectId(current_user.get("user_id"))
        has_access = (
            Permissions.PROJECT_READ_ALL in user_permissions or
            Permissions.ADMIN in user_permissions or
            project.get("is_public") or
            project.get("created_by") == user_id or
            any(member["user_id"] == user_id for member in project.get("members", []))
        )
        if not has_access:
            raise PermissionError("Not enough permissions to view this project activities")

        project_id_str = str(project["_id"])
        query = {
            "$or": [
                {"target_type": "project", "target_id": project_id_str},
                {"metadata.project_id": project_id_str}
            ]
        }

        skip = (page - 1) * size
        total = await db.activity_logs.count_documents(query)
        activities = await db.activity_logs.find(query) \
            .sort("created_at", -1) \
            .skip(skip) \
            .limit(size) \
            .to_list(length=None)

        user_id_set = set()
        for activity in activities:
            user_val = activity.get("user_id")
            if user_val:
                user_id_set.add(user_val)

        user_map: Dict[str, Any] = {}
        if user_id_set:
            user_object_ids = []
            for uid in user_id_set:
                try:
                    user_object_ids.append(ObjectId(uid))
                except Exception:
                    continue
            if user_object_ids:
                users = await db.users.find({"_id": {"$in": user_object_ids}}).to_list(length=None)
                for user in users:
                    user_map[str(user["_id"])] = user

        items: List[Dict[str, Any]] = []
        for activity in activities:
            user_key = str(activity.get("user_id")) if activity.get("user_id") else None
            user_info = user_map.get(user_key) if user_key else None
            metadata = activity.get("metadata", {}) or {}
            items.append({
                "id": str(activity.get("_id")),
                "user_id": user_key,
                "user_name": (user_info.get("name") if user_info else None) or \
                              (user_info.get("username") if user_info else None) or \
                              metadata.get("user_name") or "未知用户",
                "action": activity.get("action", ""),
                "target_type": activity.get("target_type"),
                "target_id": activity.get("target_id"),
                "target_name": activity.get("target_name"),
                "description": activity.get("description", ""),
                "type": metadata.get("type") or activity.get("target_type", "project"),
                "metadata": metadata,
                "created_at": activity.get("created_at"),
            })

        pages = (total + size - 1) // size if size > 0 else 0
        pagination = {
            "total": total,
            "page": page,
            "size": size,
            "pages": pages,
            "has_prev": page > 1,
            "has_next": page < pages
        }

        return {
            "items": items,
            "pagination": pagination
        }

    def _format_project_response(
        self,
        project: Dict[str, Any],
        task_count: int = 0,
        completed_task_count: int = 0,
        progress: float = 0.0,
        extra_counts: Optional[Dict[str, Any]] = None
    ) -> Dict[str, Any]:
        """
        格式化项目响应数据
        
        Args:
            project: 项目文档
            task_count: 任务总数
            completed_task_count: 已完成任务数
            progress: 进度百分比
            
        Returns:
            格式化后的项目数据
        """
        response = {
            "id": str(project["_id"]),
            "name": project["name"],
            "description": project.get("description"),
            "category": project["category"],
            "priority": project["priority"],
            "status": project["status"],
            "start_date": project.get("start_date"),
            "end_date": project.get("end_date"),
            "estimated_hours": project.get("estimated_hours"),
            "budget": project.get("budget"),
            "tags": project.get("tags", []),
            "is_public": project["is_public"],
            "organization_id": str(project["organization_id"]) if project.get("organization_id") else None,
            "members": [
                {
                    "user_id": str(member["user_id"]),
                    "role": member["role"],
                    "joined_at": member["joined_at"],
                    "permissions": member.get("permissions", [])
                }
                for member in project.get("members", [])
            ],
            "created_by": str(project["created_by"]),
            "created_at": project["created_at"],
            "updated_at": project["updated_at"],
            "is_archived": project.get("is_archived", False),
            "archived_at": project.get("archived_at"),
            "task_count": task_count,
            "completed_task_count": completed_task_count,
            "member_count": len(project.get("members", [])),
            "progress": progress
        }

        owner_member = next(
            (member for member in project.get("members", []) if member.get("role") == ProjectMemberRole.OWNER),
            None
        )

        if owner_member:
            response["owner_id"] = str(owner_member.get("user_id"))
        else:
            response["owner_id"] = str(project.get("created_by"))

        if extra_counts:
            response.update(extra_counts)
        
        # 确保 created_by_name 字段存在且不为空
        if "created_by_name" not in response or not response.get("created_by_name"):
            # 如果 extra_counts 中没有提供，尝试从 created_by 获取
            created_by_id = response.get("created_by")
            if created_by_id:
                # 这里可以再次查询，但为了性能，我们使用默认值
                response["created_by_name"] = "未知用户"
            else:
                response["created_by_name"] = "未知用户"

        return response


# 全局项目服务实例
project_service = ProjectService()

